package com.rocketmq.transaction.execult;

import com.rocketmq.exception.BusinessImplNullException;
import com.rocketmq.transaction.listener.DefaultTransactionListener;
import com.rocketmq.transaction.service.DispatchService;
import com.rocketmq.transaction.service.TransactionService;
import com.rocketmq.JmsProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.*;

/**
 *  事务发送生产者
 * @author: hw
 * @date: 2021/3/8 16:06
 */
@Slf4j
@Configuration
public class TransactionExecute implements RocketProducer, DispatchService {


     @Autowired private  JmsProperties jmsProperties;

     private TransactionService transactionService;

    @Override
    public SendResult transactionExecute(Message message){
        // 事务创建者组
        TransactionMQProducer producer = new TransactionMQProducer(jmsProperties.getTransactionProducerGroup());
        // 添加服务器地址
        producer.setNamesrvAddr(jmsProperties.getServerAddr());
        // 添加事务监听器
        if (null == transactionService){
            throw new BusinessImplNullException("transaction concrete realization is not null");
        }
        producer.setTransactionListener(new DefaultTransactionListener(transactionService));

        /**
         * 自定义线程池来处理检查请求
         */
        ExecutorService executorService = new ThreadPoolExecutor(
                2,
                5,
                100,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2000));

        // 添加自定义线程池
        producer.setExecutorService(executorService);
        try {
            producer.start();
            /**
             *  发送半消息体
             */
            SendResult sendResult = producer.sendMessageInTransaction(message,null);

            producer.shutdown();
            log.info("发送结果：{}, 发送返回：{}", sendResult.getSendStatus(), sendResult);
            return sendResult;
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 指定具体业务对象
     * @param transactionService  具体serviceImpl实现累
     * @return
     */
    @Override
    public TransactionService setTransactionService(TransactionService transactionService) {
        return this.transactionService = transactionService;
    }

}
